package org.shj.spark.secondsort;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;

public class SecondSort {

	public static void main(String[] args) {
		SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("SecondSort");
		JavaSparkContext sc = new JavaSparkContext(conf);
		JavaRDD<String> lines = sc.textFile("E:/workspace/scala/sparkjava/src/main/resources/sort.txt");
		
		JavaPairRDD<SecondSortKey, String> pairs = lines.mapToPair(new PairFunction<String, SecondSortKey, String>(){
			private static final long serialVersionUID = -3846886275181885786L;

			@Override
			public Tuple2<SecondSortKey, String> call(String line) throws Exception {
				String[] lineSplited = line.split("\\s+");
				SecondSortKey key = new SecondSortKey(Integer.valueOf(lineSplited[0]), 
						Integer.valueOf(lineSplited[2]));
				return new Tuple2<SecondSortKey, String>(key, line);
			}
		});
		
		JavaPairRDD<SecondSortKey, String> sortedPairs = pairs.sortByKey(false);
		
		sortedPairs.map(new Function<Tuple2<SecondSortKey, String>, String>(){
			private static final long serialVersionUID = -549532971253199871L;

			@Override
			public String call(Tuple2<SecondSortKey, String> tuple) throws Exception {
				return tuple._2;
			}
		}).foreach(new VoidFunction<String>(){
			private static final long serialVersionUID = -6455445868640848916L;

			public void call(String line) throws Exception {
				System.out.println(line);
			}
			
		});
		
		sc.close();

	}

}
